/**
 * 
 */
package skewreduce.framework.planner;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Set;
import java.util.SortedSet;

import org.apache.hadoop.conf.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import skewreduce.framework.Schedule;


public class PriorityScheduler {
	private static final Logger LOG = LoggerFactory.getLogger(PriorityScheduler.class);
	
	Set<PartitionNode> notReady = new HashSet<PartitionNode>(); 
	PriorityQueue<PartitionNode> readyQueue; // in order of job cost, category (WORK,MERGE)
	PriorityQueue<PartitionNode> scheduleQueue; // we pop this up by end time. capacity = # of machines.
	List<PartitionNode> schedule = new ArrayList<PartitionNode>();
	int numMachines;

	double scheduleDelay; // delay until start
	double currentTime;
	
	public PriorityScheduler(Configuration conf) {
		readyQueue = new PriorityQueue<PartitionNode>();
		scheduleQueue = new PriorityQueue<PartitionNode>(32,new Comparator<PartitionNode>() {
			@Override
			public int compare(PartitionNode o1, PartitionNode o2) {
				return (int) Math.signum(o1.getEndAt() - o2.getEndAt());
			}
		});
		
		scheduleDelay = conf.getFloat("skewreduce.planner.scheduler.delay",10); // in seconds
	}
	
	public double getScheduleOverhead() { return scheduleDelay; }
	public void setNumSlots(int n) {
		numMachines = n;
	}
	
	public void reset(Collection<PartitionNode> nodes) {
		currentTime = 0.0;
		notReady.clear();
		//readyQueue = new PriorityQueue<PartitionNode>(nodes);
		readyQueue = new PriorityQueue<PartitionNode>();
		for ( PartitionNode node : nodes ) {
			add(node);
		}
		scheduleQueue.clear();
		schedule.clear();
	}
	
	public void add(PartitionNode node) {
		if ( node.isLeaf() ) {
			readyQueue.add(node);
		} else {
			node.resetStatus();
			notReady.add(node);
		}
	}
	
	public double estimateSchedule() {
		if ( LOG.isDebugEnabled() ) {
			LOG.debug("begin estimating scheduling cost");
		}
		
		while ( ! notReady.isEmpty() || ! readyQueue.isEmpty() || ! scheduleQueue.isEmpty() ) {
			int avail = numMachines - scheduleQueue.size();
			while ( avail > 0 && ! readyQueue.isEmpty() ) {
				// we have available slot. schedule.
				PartitionNode node = readyQueue.poll();
				node.schedule(currentTime,scheduleDelay);
				scheduleQueue.add(node);
				schedule.add(node);
				--avail;
				if ( LOG.isDebugEnabled() ) {
					LOG.debug("{}: schedule {}",currentTime,node.getPartition());
				}
			}
			
			// b. have no available slot. pop the earliest.
			// now we have no available slot.
			{
				PartitionNode node;
				do {
					node = scheduleQueue.poll();
					currentTime = node.getEndAt();
					node.complete();
					
					PartitionNode parent = node.getParent();
					if ( LOG.isDebugEnabled() ) {
						LOG.debug("{}: complete {}",currentTime,node.getPartition());
						LOG.debug("parent = {}/notready = {}/isready = {}",new Object[] { parent,notReady.contains(parent),( parent == null ) ? false : parent.isReady() });
						
					}
					if ( parent != null && notReady.contains(parent) && parent.isReady() ) {
						readyQueue.add(parent);
						notReady.remove(parent);
					}
				} while ( ! scheduleQueue.isEmpty() && scheduleQueue.peek().getEndAt() <= currentTime );
			}
		}
		
		if ( LOG.isDebugEnabled() ) {
			LOG.debug("estimated cost = {}",currentTime);
		}
		
		return currentTime;
	}
	
	public List<Schedule> getSchedule() {
		ArrayList<Schedule> ret = new ArrayList<Schedule>(schedule.size());
		for ( PartitionNode node : schedule ) {
			Schedule entry = new Schedule( node.getOpID(), node.getSecs(), node.getCost() );
			ret.add(entry);
		}
		return ret;
	}
}